package com.atguigu.flink.timeAndwindow;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import com.atguigu.flink.pojo.UrlViewCount;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Created by 黄凯 on 2023/6/18 0018 15:29
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 统计url的点击次数
 * *    通过增量聚合和全窗口函数的结合来实现。
 */
public class Flink09_UrlViewCount {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> ds = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );
        ds.print("input");

        //统计每10秒url的点击次数
        ds.map(
                event -> Tuple2.of(event.getUrl(), 1)
        ).returns(
                Types.TUPLE(Types.STRING, Types.INT)
        ).keyBy(
                t -> t.f0
        ).window(
                TumblingEventTimeWindows.of(Time.seconds(10))
        ).aggregate(
                new AggregateFunction<Tuple2<String, Integer>, Tuple2<String, Long>, UrlViewCount>() {
                    @Override
                    public Tuple2<String, Long> createAccumulator() {
                        return Tuple2.of("", 0L);
                    }

                    @Override
                    public Tuple2<String, Long> add(Tuple2<String, Integer> value, Tuple2<String, Long> accumulator) {
                        accumulator.setFields(value.f0, accumulator.f1 + 1);
                        return accumulator;
                    }

                    @Override
                    public UrlViewCount getResult(Tuple2<String, Long> accumulator) {

                        UrlViewCount urlViewCount = new UrlViewCount();
                        urlViewCount.setUrl(accumulator.f0);
                        urlViewCount.setCount(accumulator.f1);

                        return urlViewCount;
                    }

                    @Override
                    public Tuple2<String, Long> merge(Tuple2<String, Long> a, Tuple2<String, Long> b) {
                        return null;
                    }
                },

                new ProcessWindowFunction<UrlViewCount, UrlViewCount, String, TimeWindow>() {


                    @Override
                    public void process(String key, ProcessWindowFunction<UrlViewCount, UrlViewCount, String, TimeWindow>.Context context,
                                        Iterable<UrlViewCount> elements, Collector<UrlViewCount> out) throws Exception {

                        //获取窗口中的数据
                        //窗口中的数据已经被增量聚合函数聚合成一个结果， 所以, 全窗口函数中的elements中只会有一条数据。
                        UrlViewCount urlViewCount = elements.iterator().next();

                        long start = context.window().getStart();
                        long end = context.window().getEnd();

                        urlViewCount.setStart(start);
                        urlViewCount.setEnd(end);

                        out.collect(urlViewCount);

                    }
                }


        ).print();

        env.execute();

    }


}
